package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo3DataSource {
  def main(args: Array[String]): Unit = {
    //1、创建Spark sql环境
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("sql")
      .config("spark.sql.shuffle.partitions", 1) //默认在集群中时200个
      .getOrCreate()

    /**
     * csv : 文本文件
     */

    //读取csv格式
    val users: DataFrame = spark
      .read
      .option("sep", "|") //指定分隔符
      .schema("id STRING,name STRING ,age INT") //指定列名和类型
      .csv("spark/data/users.csv")

    //将结果保存为csv格式
    users
      .write
      .option("sep", "\t") //指定分隔符
      .mode(SaveMode.Overwrite)
      .csv("spark/data/csv")

    /**
     * json
     * json格式中自带了列名和类型，占用的空间会更多，所以在做大数据存储时不采用json格式
     */

    //读取json格式
    //会自动识别列名和类型
    val studentJSON: DataFrame = spark
      .read
      .json("spark/data/students.json")

    //打印DF的表结构
    studentJSON.printSchema()
    studentJSON.show()

    import org.apache.spark.sql.functions._
    val clazzNum: DataFrame = studentJSON
      .groupBy("clazz")
      .agg(count("id") as "num")

    //将结果保存为json格式
    clazzNum
      .write
      .mode(SaveMode.Overwrite)
      .json("spark/data/json")

    /**
     * parquet
     * 1、自带列名和类型
     * 2、会对数据进行压缩
     */

    //将结果保存到为parquet
    studentJSON.write
      .mode(SaveMode.Overwrite)
      .parquet("spark/data/parquet")


    //读取parquet格式
    val studentParquet: DataFrame = spark
      .read.parquet("spark/data/parquet")

    studentParquet.show()
  }
}
